package com.meiya.flink.config;

import java.util.Properties;
import java.util.UUID;

import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 应用配置类，集中管理所有配置参数
 */
public class AppConfig {
    private static final Logger logger = LoggerFactory.getLogger(AppConfig.class);

    // 环境配置开关 - true:生产环境, false:测试环境
    private static boolean PRODUCTION_ENV = true;

    /**
     * 设置环境
     * @param isProduction true表示生产环境，false表示测试环境
     */
    public static void setEnvironment(boolean isProduction) {
        PRODUCTION_ENV = isProduction;
        logger.info("当前环境设置为: {}", isProduction ? "生产环境" : "测试环境");
    }

    /**
     * 获取当前环境
     * @return true表示生产环境，false表示测试环境
     */
    public static boolean isProductionEnv() {
        return PRODUCTION_ENV;
    }

    // Kafka 配置
    // 测试环境
    private static final String TEST_KAFKA_BROKERS = "192.168.182.128:9092";
    // 生产环境
    private static final String PROD_KAFKA_BROKERS = "192.168.182.128:9092";

    private static final String KAFKA_CLIENT_ID_PREFIX = "flink-tags-processor";
    private static final String KAFKA_CONSUMER_GROUP_ID = "flink-tags-processor-group";
    private static final String KAFKA_TRANSACTION_TOPIC = "transaction-data";
    private static final String KAFKA_ACCOUNT_TOPIC = "account-data";
    private static final String KAFKA_TAGS_RESULT_TOPIC = "tags-data";

    // Doris 配置
    // 测试环境
    private static final String TEST_DORIS_HOST = "192.168.182.128";
    private static final String TEST_DORIS_MYSQL_PORT = "9030";
    private static final String TEST_DORIS_HTTP_PORT = "8030";
    // 生产环境
    private static final String PROD_DORIS_HOST = "192.168.182.128";
    private static final String PROD_DORIS_MYSQL_PORT = "9030";
    private static final String PROD_DORIS_HTTP_PORT = "8030";

    // 统一认证配置
    private static final String TEST_DORIS_USERNAME = "root";
    private static final String TEST_DORIS_PASSWORD = "doris";
    private static final String PROD_DORIS_USERNAME = "root";
    private static final String PROD_DORIS_PASSWORD = "doris";
    
    private static final String DORIS_DATABASE = "trade_db";

    // Redis 配置
    // 测试环境
    private static final String TEST_REDIS_HOST = "192.168.182.128";
    private static final int TEST_REDIS_PORT = 6379;
    private static final String TEST_REDIS_PASSWORD = "";
    // 生产环境
    private static final String PROD_REDIS_HOST = "redis-cluster.meiya.com";
    private static final int PROD_REDIS_PORT = 6379;
    private static final String PROD_REDIS_PASSWORD = "R3d!sPr0d@2025";

    // 批处理阈值配置
    private static final int TEST_LARGE_BATCH_THRESHOLD = 100000;
    private static final int PROD_LARGE_BATCH_THRESHOLD = 500000;

    // 弹性时间间隔配置（分钟）
    // 用于确保 account-data 和 transaction-data 两个kafka的数据源的数据一致性
    // 因为网络等原因，可能kafka消息不是同一时间到达，所以需要预留出弹性时间间隔
    // 确保同一时间段内的消息能够完全地被消费，不论 account-data 先到达，还是 transaction-data 先到达
    private static int TEST_SESSION_WINDOW_INTERVAL_MINUTES = 3;
    private static int PROD_SESSION_WINDOW_INTERVAL_MINUTES = 5;
    
    /**
     * 获取会话窗口间隔时间（分钟）
     */
    public static int getSessionWindowIntervalMinutes() {
        return PRODUCTION_ENV ? PROD_SESSION_WINDOW_INTERVAL_MINUTES : TEST_SESSION_WINDOW_INTERVAL_MINUTES;
    }
    
    /**
     * 设置会话窗口间隔时间（分钟）
     * 
     * @param minutes 间隔时间（分钟）
     */
    public static void setSessionWindowIntervalMinutes(int minutes) {
        if (minutes > 0) {
            if (PRODUCTION_ENV) {
                PROD_SESSION_WINDOW_INTERVAL_MINUTES = minutes;
                logger.info("生产环境会话窗口间隔时间已设置为: {} 分钟", PROD_SESSION_WINDOW_INTERVAL_MINUTES);
            } else {
                TEST_SESSION_WINDOW_INTERVAL_MINUTES = minutes;
                logger.info("测试环境会话窗口间隔时间已设置为: {} 分钟", TEST_SESSION_WINDOW_INTERVAL_MINUTES);
            }
        } else {
            logger.warn("会话窗口间隔时间设置无效: {}, 保持原值: {}", minutes, 
                    PRODUCTION_ENV ? PROD_SESSION_WINDOW_INTERVAL_MINUTES : TEST_SESSION_WINDOW_INTERVAL_MINUTES);
        }
    }

    /**
     * 获取Kafka连接地址
     */
    public static String getKafkaBrokers() {
        return PRODUCTION_ENV ? PROD_KAFKA_BROKERS : TEST_KAFKA_BROKERS;
    }

    /**
     * 获取Kafka客户端ID前缀
     */
    public static String getKafkaClientIdPrefix() {
        return KAFKA_CLIENT_ID_PREFIX;
    }

    /**
     * 获取Kafka消费者组ID
     */
    public static String getKafkaConsumerGroupId() {
        return KAFKA_CONSUMER_GROUP_ID;
    }

    /**
     * 获取Kafka交易数据主题
     */
    public static String getKafkaTransactionTopic() {
        return KAFKA_TRANSACTION_TOPIC;
    }

    /**
     * 获取Kafka账户数据主题
     */
    public static String getKafkaAccountTopic() {
        return KAFKA_ACCOUNT_TOPIC;
    }

    /**
     * 获取Kafka标签结果主题
     */
    public static String getKafkaTagsResultTopic() {
        return KAFKA_TAGS_RESULT_TOPIC;
    }

    /**
     * 获取Doris主机地址
     */
    public static String getDorisHost() {
        return PRODUCTION_ENV ? PROD_DORIS_HOST : TEST_DORIS_HOST;
    }

    /**
     * 获取Doris MySQL节点地址
     */
    public static String getDorisMysqlNodes() {
        return getDorisHost() + ":" + (PRODUCTION_ENV ? PROD_DORIS_MYSQL_PORT : TEST_DORIS_MYSQL_PORT);
    }

    /**
     * 获取Doris HTTP节点地址
     */
    public static String getDorisHttpNodes() {
        return getDorisHost() + ":" + (PRODUCTION_ENV ? PROD_DORIS_HTTP_PORT : TEST_DORIS_HTTP_PORT);
    }

    /**
     * 获取Doris用户名
     */
    public static String getDorisUsername() {
        return PRODUCTION_ENV ? PROD_DORIS_USERNAME : TEST_DORIS_USERNAME;
    }

    /**
     * 获取Doris密码
     */
    public static String getDorisPassword() {
        return PRODUCTION_ENV ? PROD_DORIS_PASSWORD : TEST_DORIS_PASSWORD;
    }

    /**
     * 获取Doris数据库名
     */
    public static String getDorisDatabase() {
        return DORIS_DATABASE;
    }

    /**
     * 获取Redis主机地址
     */
    public static String getRedisHost() {
        return PRODUCTION_ENV ? PROD_REDIS_HOST : TEST_REDIS_HOST;
    }

    /**
     * 获取Redis端口
     */
    public static int getRedisPort() {
        return PRODUCTION_ENV ? PROD_REDIS_PORT : TEST_REDIS_PORT;
    }

    /**
     * 获取Redis密码
     */
    public static String getRedisPassword() {
        return PRODUCTION_ENV ? PROD_REDIS_PASSWORD : TEST_REDIS_PASSWORD;
    }

    /**
     * 获取大批量处理阈值
     */
    public static int getLargeBatchThreshold() {
        return PRODUCTION_ENV ? PROD_LARGE_BATCH_THRESHOLD : TEST_LARGE_BATCH_THRESHOLD;
    }

    /**
     * 获取Kafka生产者配置
     */
    public static Properties getKafkaProducerProperties() {
        Properties props = new Properties();
        props.put("bootstrap.servers", getKafkaBrokers());
        props.put("client.id", getKafkaClientIdPrefix() + "-producer-" + UUID.randomUUID());
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        
        // 增加最大请求大小配置，默认是1MB，这里设置为8MB
        props.put("max.request.size", "8388608");  // 8MB
        props.put("buffer.memory", "67108864");    // 64MB
        props.put("batch.size", "131072");         // 128KB
        props.put("linger.ms", "10");              // 批量发送等待时间
        props.put("compression.type", "snappy");   // 启用压缩
        
        // 生产环境添加安全认证配置
        if (PRODUCTION_ENV) {
            // 生产环境特定参数
            props.put("acks", "all");             // 确保数据持久化
            props.put("retries", "5");            // 重试次数
            props.put("retry.backoff.ms", "500"); // 重试间隔
        } else {
            // 测试环境简化配置
            props.put("acks", "1");
            props.put("retries", "3");
            props.put("retry.backoff.ms", "100");
        }
        
        return props;
    }

    /**
     * 获取Kafka管理客户端配置
     */
    public static Properties getKafkaAdminProperties() {
        Properties props = new Properties();
        props.put("bootstrap.servers", getKafkaBrokers());
        props.put("client.id", getKafkaClientIdPrefix() + "-admin-" + UUID.randomUUID());
        props.put("group.id", getKafkaConsumerGroupId());
        
        // 生产环境添加安全认证配置
        if (PRODUCTION_ENV) {
            // 配置SASL认证 - 生产环境
            props.put("security.protocol", "SASL_PLAINTEXT");
            props.put("sasl.mechanism", "PLAIN");
            props.put("sasl.jaas.config", 
                    "org.apache.kafka.common.security.plain.PlainLoginModule required " +
                    "username=\"kafka_prod_user\" " +
                    "password=\"kafka_prod_password\";");
        }
        
        return props;
    }

    /**
     * 获取Kafka消费者配置
     */
    public static Properties getKafkaConsumerProperties() {
        Properties props = new Properties();
        props.put("bootstrap.servers", getKafkaBrokers());
        props.put("group.id", getKafkaConsumerGroupId());
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("auto.offset.reset", "latest");
        
        // 消费者配置 - 生产环境特定配置
        if (PRODUCTION_ENV) {
            // 增加安全配置
            props.put("security.protocol", "SASL_PLAINTEXT");
            props.put("sasl.mechanism", "PLAIN");
            props.put("sasl.jaas.config", 
                    "org.apache.kafka.common.security.plain.PlainLoginModule required " +
                    "username=\"kafka_prod_user\" " +
                    "password=\"kafka_prod_password\";");
            
            // 调整生产环境消费性能参数
            props.put("fetch.min.bytes", "65536");  // 64KB
            props.put("fetch.max.wait.ms", "500");
            props.put("max.partition.fetch.bytes", "1048576");  // 1MB
            props.put("max.poll.records", "1000");
            props.put("heartbeat.interval.ms", "3000");
            props.put("session.timeout.ms", "45000");
            props.put("max.poll.interval.ms", "300000");  // 5分钟
        } else {
            // 测试环境简化配置
            props.put("fetch.min.bytes", "1");
            props.put("fetch.max.wait.ms", "100");
            props.put("max.poll.records", "500");
            props.put("heartbeat.interval.ms", "3000");
            props.put("session.timeout.ms", "30000");
            props.put("request.timeout.ms", "40000");
        }
        
        return props;
    }

    /**
     * 配置TableEnvironment
     * 设置SQL查询相关的优化参数
     *
     * @param tableEnv 表环境对象
     */
    public static void configureTableEnvironment(StreamTableEnvironment tableEnv) {
        TableConfig tableConfig = tableEnv.getConfig();
        
        // 基本配置 - 两种环境通用
        tableConfig.setIdleStateRetention(java.time.Duration.ofHours(1));
        
        if (PRODUCTION_ENV) {
            // ========== 生产环境配置 ==========
            logger.info("使用生产环境配置初始化TableEnvironment");
            
            // 设置SQL操作并行度 - 生产环境更高并行度
            tableConfig.set("table.exec.resource.default-parallelism", "16");
            
            // 启用小批量处理，优化生产环境处理性能
            tableConfig.set("table.exec.mini-batch.enabled", "true");
            tableConfig.set("table.exec.mini-batch.size", "50000");  // 更大的批处理
            tableConfig.set("table.exec.mini-batch.allow-latency", "3s");
            
            // 内存优化配置 - 生产环境
            tableConfig.set("table.optimizer.distinct-agg.split.enabled", "true");
            tableConfig.set("table.optimizer.reuse-sub-plan-enabled", "true");
            tableConfig.set("table.optimizer.join-reorder-enabled", "true");

            // 查询超时设置 - 生产环境更长超时
            tableConfig.set("table.exec.query-timeout", "600000");  // 10分钟

            // 状态后端配置 - 生产使用RocksDB
            tableConfig.set("state.backend", "rocksdb");
            tableConfig.set("state.backend.incremental", "true");
            tableConfig.set("state.backend.rocksdb.memory.managed", "true");
            tableConfig.set("state.backend.rocksdb.block.cache-size", "4096mb");  // 生产环境更大缓存
            tableConfig.set("state.backend.rocksdb.write-batch-size", "8mb");
            tableConfig.set("state.backend.rocksdb.thread.num", "16");  // 更多线程
            tableConfig.set("state.backend.rocksdb.files.open", "1000");
            tableConfig.set("state.backend.local-recovery", "true");
            tableConfig.set("state.backend.rocksdb.timer-service.factory", "ROCKSDB");
            
            // 生产环境允许算子链，提高性能
            tableConfig.set("pipeline.operator-chaining.enabled", "true");

            // 检查点配置 - 生产环境配置高可用性
            tableConfig.set("execution.checkpointing.interval", "180000");  // 3分钟
            tableConfig.set("execution.checkpointing.timeout", "600000");  // 10分钟
            tableConfig.set("execution.checkpointing.max-concurrent-checkpoints", "1");
            tableConfig.set("execution.checkpointing.min-pause", "30000");  // 30秒
            tableConfig.set("execution.checkpointing.mode", "EXACTLY_ONCE");  // 生产环境强一致性
            tableConfig.set("execution.checkpointing.tolerable-failed-checkpoints", "3");  // 允许少量失败
            tableConfig.set("execution.checkpointing.cleanup-mode", "RETAIN_ON_CANCELLATION");

            // 重启策略 - 生产环境自动恢复
            tableConfig.set("restart-strategy.type", "fixed-delay");
            tableConfig.set("restart-strategy.fixed-delay.attempts", "10");  // 最多尝试10次
            tableConfig.set("restart-strategy.fixed-delay.delay", "30s");  // 30秒间隔
            
            // Doris连接配置 - 生产环境
            tableConfig.set("doris.request.connect-timeout", "300000");  // 5分钟
            tableConfig.set("doris.request.read-timeout", "300000");  // 5分钟
            tableConfig.set("doris.request.query-timeout", "600000");  // 10分钟
            tableConfig.set("doris.request.retries", "10");  // 更多重试次数
            tableConfig.set("doris.execution.wait-timeout", "1200000");  // 20分钟
            tableConfig.set("doris.table-options.request.tablet.size", "3");  // 减少tablet大小，更均衡
            tableConfig.set("doris.table-options.batch.size", "1024");  // 适中批量，避免OOM
            tableConfig.set("doris.table-options.exec.mem.limit", "17179869184");  // 16GB

            logger.info("生产环境TableEnvironment配置完成 - 高可用、高性能、强一致性");
        } else {
            // ========== 测试环境配置 ==========
            logger.info("使用测试环境配置初始化TableEnvironment");
            
            // 设置SQL操作并行度 - 测试环境较低并行度
            tableConfig.set("table.exec.resource.default-parallelism", "8");
            
            // 启用小批量处理 - 测试环境
            tableConfig.set("table.exec.mini-batch.enabled", "true");
            tableConfig.set("table.exec.mini-batch.size", "10000");
            tableConfig.set("table.exec.mini-batch.allow-latency", "5s");
            
            // 优化内存使用 - 测试环境
            tableConfig.set("table.optimizer.distinct-agg.split.enabled", "true");
            tableConfig.set("table.optimizer.reuse-sub-plan-enabled", "true");
            tableConfig.set("table.optimizer.join-reorder-enabled", "true");

            // 设置查询超时时间 - 测试环境
            tableConfig.set("table.exec.query-timeout", "360000");  // 6分钟

            // 状态后端配置 - 测试环境
            tableConfig.set("state.backend", "rocksdb");
            tableConfig.set("state.backend.incremental", "true");
            tableConfig.set("state.backend.rocksdb.memory.managed", "true");
            tableConfig.set("state.backend.rocksdb.block.cache-size", "1024mb");  // 1GB
            tableConfig.set("state.backend.rocksdb.write-batch-size", "4mb");
            tableConfig.set("state.backend.rocksdb.thread.num", "8");
            tableConfig.set("state.backend.rocksdb.files.open", "1000");
            tableConfig.set("state.backend.local-recovery", "true");
            tableConfig.set("state.backend.rocksdb.timer-service.factory", "ROCKSDB");
            
            // 禁用算子链以隔离每个算子 - 便于测试环境调试
            tableConfig.set("pipeline.operator-chaining.enabled", "false");

            // 设置检查点配置 - 测试环境便于调试
            tableConfig.set("execution.checkpointing.interval", "60000");  // 1分钟
            tableConfig.set("execution.checkpointing.timeout", "600000");  // 10分钟
            tableConfig.set("execution.checkpointing.max-concurrent-checkpoints", "1");
            tableConfig.set("execution.checkpointing.min-pause", "10000");  // 10秒
            tableConfig.set("execution.checkpointing.mode", "AT_LEAST_ONCE");  // 简化模式
            tableConfig.set("execution.checkpointing.tolerable-failed-checkpoints", "0");  // 不容忍失败
            tableConfig.set("execution.checkpointing.cleanup-mode", "RETAIN_ON_CANCELLATION");  // 保留检查点

            // 设置重启策略 - 测试环境不自动重启便于调试
            tableConfig.set("restart-strategy.type", "none");  // 不自动重启
            tableConfig.set("restart-strategy.fixed-delay.attempts", "0");
            tableConfig.set("restart-strategy.fixed-delay.delay", "0s");
            
            // 设置Doris连接配置 - 测试环境
            tableConfig.set("doris.request.connect-timeout", "180000");  // 3分钟
            tableConfig.set("doris.request.read-timeout", "180000");  // 3分钟
            tableConfig.set("doris.request.query-timeout", "360000");  // 6分钟
            tableConfig.set("doris.request.retries", "5");  // 5次重试
            tableConfig.set("doris.execution.wait-timeout", "600000");  // 10分钟
            tableConfig.set("doris.table-options.request.tablet.size", "5");
            tableConfig.set("doris.table-options.batch.size", "2048");
            tableConfig.set("doris.table-options.exec.mem.limit", "8589934592");  // 8GB

            logger.info("测试环境TableEnvironment配置完成 - 已关闭自动重启策略，隔离算子链，用于问题排查");
        }
    }
}